%% copy riak-0.14.2 source to learn dynamo implementation and erlang
%% author: lichuang
%% copy source from riak_kv_delete.erl
%% riak_delete: two-step object deletion
-module(edynamo_delete).
-export([delete/6]).

%% @spec delete(ReqId :: binary(), edynamo_object:bucket(), edynamo_object:key(),
%%             RW :: integer(), TimeoutMillisecs :: integer(), Client :: pid())
%%           -> term()
%% @doc Delete the object at Bucket/Key.  Direct return value is uninteresting,
%%      see edynamo_client:delete/3 for expected gen_server replies to Client.
delete(ReqId, Bucket, Key, RW0, Timeout, Client) ->
  RealStartTime = edynamo_util:moment(),
  {ok, Ring} = edynamo_ring_manager:get_my_ring(),
  BucketProps = edynamo_bucket:get_bucket(Bucket, Ring),
  N = proplists:get_value(n_val, BucketProps),
  RW = edynamo_util:expand_rw_value(rw, RW0, BucketProps, N),
  {ok, C} = edynamo_util:local_client(),
  case C:get(Bucket, Key, RW, Timeout) of
    {ok, OrigObj} ->
      RemainingTime = Timeout - (edynamo_util:moment() - RealStartTime),
      OrigMD = hd([MD || {MD, _V} <- edynamo_object:get_contents(OrigObj)]),
      NewObj = edynamo_object:update_metadata(OrigObj,
		  dict:store(<<"X-Riak-Deleted">>, "true", OrigMD)),
      Reply = C:put(NewObj, RW, RW, RemainingTime),
      case Reply of
	ok ->
	  spawn(
	    fun() -> reap(Bucket, Key, RemainingTime) end);
	_ -> nop
      end,
      Client ! {ReqId, Reply};
    {error, notfound} ->
      Client ! {ReqId, {error, notfound}};
    X ->
      Client ! {ReqId, X}
  end.

reap(Bucket, Key, Timeout) ->
  {ok, C} = edynamo_util:local_client(),
  {ok, Ring} = edynamo_ring_manager:get_my_ring(),
  BucketProps = edynamo_bucket:get_bucket(Bucket, Ring),
  N = proplists:get_value(n_val, BucketProps),
  C:get(Bucket, Key, N, Timeout).
